package com.iwunu.gateway.listener;

import java.util.HashMap;
import java.util.Map;

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import com.iwunu.gateway.callback.TopMsgCallback;
import com.iwunu.gateway.config.MqttCofigBean;
import com.iwunu.gateway.service.FLcokMqttReceiveService;

@Component
public class LockMqttListener implements ApplicationRunner{
	private static Logger logger= LoggerFactory.getLogger(LockMqttListener.class);
	private static MqttClient client;
	private static MqttTopic mqttTopic;
	@Autowired
	private FLcokMqttReceiveService fLcokMqttReceiveService;
	/**
     * MQTT连接属性配置对象
     */
    @Autowired
    public MqttCofigBean mqttCofigBean;
    /**
     * 初始化参数配置
     */
    @Override
    public void run(ApplicationArguments args) throws Exception {
        logger.info("初始化启动MQTT连接");
        this.connect();
    }


    /**
     *  用来连接服务器
     */
    private void connect() throws Exception{
        client = new MqttClient(mqttCofigBean.getHostUrl(),mqttCofigBean.getSclientId(), new MemoryPersistence());
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(mqttCofigBean.getUsername());
        options.setPassword(mqttCofigBean.getPassword().toCharArray());
        //是否清除session
        options.setCleanSession(false);
        // 设置超时时间
        options.setConnectionTimeout(mqttCofigBean.getCompletionTimeout());
        options.setAutomaticReconnect(true);
        // 设置会话心跳时间
        options.setKeepAliveInterval(mqttCofigBean.getKeepalive());
        
        try {
            String[] msgtopic = mqttCofigBean.getReceiveTopic().split(",");
            String[] sendtopic = mqttCofigBean.getSendTopic().split(",");
            Map<String, String> topicMap = new HashMap<String, String>();
            //订阅消息
            int[] qos = new int[msgtopic.length];
            for (int i = 0; i < msgtopic.length; i++) {
                qos[i]=0;
            }
            if(sendtopic!=null&&sendtopic.length>0) {
            	
            	for(int i=0;i<sendtopic.length;i++) {
            		String key = sendtopic[i].substring(sendtopic[i].lastIndexOf(".")+1, sendtopic[i].length());
            		topicMap.put(key, sendtopic[i]);
            	}
            }
            client.setCallback(new TopMsgCallback(client,options,sendtopic,qos,fLcokMqttReceiveService,topicMap,mqttCofigBean));
            client.connect(options);
            client.subscribe(msgtopic,qos);
            logger.info("MQTT连接成功:"+mqttCofigBean.getSclientId()+":"+client);
        } catch (Exception e) {
            logger.error("MQTT连接异常："+e);
        }
    }


    /**
     * 重连
     * @throws Exception
     */
    public void reConnect() throws Exception {
    	logger.error("MQTT+"+mqttCofigBean.getSclientId()+"+ Reconnect!!!");
        long reconnectTimes = 1;
        while (true) {
            try {
                if (client.isConnected()) {
                	logger.error("mqtt reconnect success end");
                    break;
                }
                if(reconnectTimes == 10){
                    //当重连次数达到10次时，就抛出异常，不在重连
                	logger.warn("mqtt reconnect error");
                    return;
                }
                logger.warn("mqtt reconnect times = {} try again...", reconnectTimes++);
                client.reconnect();
            } catch (MqttException e) {
            	logger.error("", e);
            }
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e1) {
                e1.printStackTrace();
            }
        }
    }

    /**
     * 订阅某个主题
     * @param topic
     * @param qos
     */
    public void subscribe(String[] receiveTopic,int[] qos){
        try {
            logger.info("receiveTopic:"+receiveTopic);
            client.subscribe(receiveTopic, qos);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    public MqttClient getClient() {
        return client;
    }

    public void setClient(MqttClient client) {
        this.client = client;
    }

    public MqttTopic getMqttTopic() {
        return mqttTopic;
    }

    public void setMqttTopic(MqttTopic mqttTopic) {
        this.mqttTopic = mqttTopic;
    }
}
